package org.fujene.communicate.repli;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.lang.reflect.Constructor;
import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.logging.FileHandler;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.logging.SimpleFormatter;

import org.fujene.index.ReplicateIndexSystem;
import org.fujene.memman.ContentSnapshoter;
import org.fujene.memman.InvertIndexSnapshoter;
import org.fujene.memman.ReplicateMemoryManager;
import org.fujene.memman.TermSnapshoter;
import org.fujene.search.ReplicateSearchSystem;
import org.fujene.structure.Constants;
import org.fujene.structure.ContentInfo;
import org.fujene.structure.ExtendQueue;
import org.fujene.structure.ExtendResult;
import org.fujene.structure.ExtendTerm;
import org.fujene.structure.HashIDExt;
import org.fujene.structure.LookupPointer;
import org.fujene.structure.Node;
import org.fujene.structure.ReplicateInfo;
import org.fujene.structure.SearchQueue;
import org.fujene.structure.Term;
import org.fujene.structure.TermList;
import org.fujene.term.Chopper;
import org.fujene.term.SenChopper;
import org.fujene.toolbox.ContentHash;
import org.msgpack.MessagePackObject;
import org.msgpack.object.ArrayType;
import org.msgpack.object.IntegerType;
import org.msgpack.object.RawType;
import org.msgpack.rpc.Server;
import org.msgpack.rpc.loop.EventLoop;

public class ReplicatePeerSystem{
	public static final int OUTER = 0;
	public static final int INNER = 1;

	// MemoryManager -> Memory, Disk
	private ReplicateMemoryManager memman = null;

	public boolean stopFlag = false;

	private ExecutorService execIndex = null;
	private ExecutorService execSearch = null;
	private ExecutorService execNodeLive = null;
	private Thread liveThreadOuter = null;
	private Thread liveThreadInner = null;

	private String directory = null;
	private int blockNum = 0;
	private int blockSize = 0;

	// //////// P2P ///////////
	private static final int PEERPORT = Constants.PEER_PORT;

	private int replication;
	private int indexType;
	private int retryCount;

	private TreeSet<Node> termNodes = null;
	private TreeSet<Node> contentNodes = null;
	private Node myTermNode = null;
	private Node myContentNode = null;
	private String[] firstHost = null;
	private Server server = null;
	private Chopper chopper = null;
	private ContentInfo contentInfo = null;
	private ReplicateCommunicator communicator = null;

	private boolean isRerank = false;
	// //////// Logger ///////////
	private Logger mainLogger = null;

	public void initialize() throws IOException, InterruptedException{
		communicator = new ReplicateCommunicator();

		termNodes = new TreeSet<Node>();
		contentNodes = new TreeSet<Node>();

		// 自分のIPアドレス取得
		Enumeration<NetworkInterface> enulfs = NetworkInterface.getNetworkInterfaces();
		if(!enulfs.hasMoreElements()) return;
		NetworkInterface ni = enulfs.nextElement();

		Enumeration<InetAddress> enuAddr = ni.getInetAddresses();
		if(!enuAddr.hasMoreElements()) return;
		// FIXME: 設定でIPv4/IPv6を選べるようにする 課題点？
		InetAddress addrInfo = enuAddr.nextElement();
		while(!(addrInfo instanceof Inet4Address))
			addrInfo = enuAddr.nextElement();

		long[] hash = calcHash(addrInfo);
		String address = addrInfo.getHostAddress();
		myTermNode = new Node(address, hash[0]);
		termNodes.add(myTermNode);
		myContentNode = new Node(address, hash[1]);
		contentNodes.add(myContentNode);
		getInformation();

		// 初期値
		if(contentInfo == null) contentInfo = new ContentInfo(new String[]{ "content" }, new String[0]);

		server = new Server(EventLoop.defaultEventLoop());
		server.serve(new ReplicatePeerHandler(mainLogger, this, contentInfo.getContentNum(), contentInfo
				.getAppendixNum()));
		server.listen(PEERPORT);

		new ReplicateSearchSystem(this, mainLogger);
		new ReplicateIndexSystem(this, contentInfo.getContentNum(), contentInfo.getAppendixNum(), mainLogger);

		memman = new ReplicateMemoryManager(blockSize, blockNum, directory, mainLogger, contentInfo);

		execIndex = Executors.newFixedThreadPool(1);
		execSearch = Executors.newFixedThreadPool(1);
		execNodeLive = Executors.newFixedThreadPool(2);

		chopper = new SenChopper();
		liveThreadOuter = new SurvivingService(OUTER);
		execNodeLive.execute(liveThreadOuter);
		liveThreadInner = new SurvivingService(INNER);
		execNodeLive.execute(liveThreadInner);
		if(firstHost != null){
			registerNode();
			callForData(OUTER);
			callForData(INNER);
		}else iamIndexable();
	}

	private boolean getArgs(String[] args){
		boolean isEnded = false;
		String settingFile;

		if(args.length > 2 || args.length == 1){
			System.err.println("Usage: PeerSystem [--setting <SettingFile>] [--secondary <IPAddress ...>]");
			return isEnded;
		}
		// デフォルト値
		this.blockNum = 256;
		this.blockSize = 1048576;
		this.directory = "./fujene_data/";
		this.indexType = 1;
		this.replication = 1;
		this.retryCount = 2;
		this.mainLogger = Logger.getLogger("Fujene search engine");
		try{
			FileHandler handle = new FileHandler("fujene_data/fujene.log");
			handle.setFormatter(new SimpleFormatter());
			this.mainLogger.addHandler(handle);
		}catch(Exception e1){
			e1.printStackTrace();
		}
		this.mainLogger.setLevel(Level.ALL);

		// ローカル
		ArrayList<String> contentName = new ArrayList<String>();
		ArrayList<String> supplementName = new ArrayList<String>();
		int count = 0;

		if(args.length > 1 && args[0].equals("--setting")){
			settingFile = args[1];

			BufferedReader br = null;
			try{
				br = new BufferedReader(new FileReader(new File(settingFile)));

				// properties
				Properties props = new Properties();
				String propval = null;
				props.load(br);

				// "#"始まりはコメント行
				if((propval = props.getProperty("BlockNum")) != null) blockNum = Integer.parseInt(propval);
				if((propval = props.getProperty("BlockSize")) != null) blockSize = Integer.parseInt(propval);
				if((propval = props.getProperty("Directory")) != null) directory = propval;
				if((propval = props.getProperty("PeerMode")) != null)
					if(propval.equalsIgnoreCase("on")) replication = 3;
				if((propval = props.getProperty("replication")) != null)
					replication = Integer.parseInt(propval);
				if((propval = props.getProperty("IndexMode")) != null) indexType = Integer.parseInt(propval);
				if((propval = props.getProperty("LogLevel")) != null){
					try{
						mainLogger.setLevel(Level.parse(propval));
					}catch(Exception e){
						mainLogger.severe("Specified log-level does not exist.");
					}
				}
				if((propval = props.getProperty("Chopper")) != null){
					try{
						@SuppressWarnings("unchecked") Class<Chopper> cls = (Class<Chopper>)Class
								.forName(propval);
						Constructor<Chopper> construct = cls.getConstructor();
						chopper = construct.newInstance();
					}catch(ClassNotFoundException e){
						System.out.println("Class for chopper did not found. application will be shut down.");
					}catch(Exception e){
						throw new InternalError("Internal error, probably a bug.");
					}
				}
				if((propval = props.getProperty("Content")) != null) contentName.add(propval);
				if((propval = props.getProperty("Supplement")) != null) supplementName.add(propval);
				if((propval = props.getProperty("Rerank")) != null)
					if(propval.equalsIgnoreCase("TFIDF")) isRerank = true;
				if((propval = props.getProperty("Connection")) != null){
					// この時点で紐付け情報を作る
					if(contentInfo == null)
						contentInfo = new ContentInfo(contentName.toArray(new String[0]),
								supplementName.toArray(new String[0]));

					String[] values = propval.split(",");
					int[] indices = new int[values.length];
					for(int i = 0; i < indices.length; i++)
						indices[i] = Integer.parseInt(values[i]);
					contentInfo.connectGraph(count++, indices);
				}
				isEnded = true;
			}catch(FileNotFoundException e){
				mainLogger.severe("Please input appropriate file. Search system will shutdown.");
				return isEnded;
			}catch(IOException e){}finally{
				if(br != null){
					try{
						br.close();
					}catch(IOException e){}
				}
			}
		}else if(args.length == 0){
			isEnded = true;
		}

		// セカンダリモード
		if(args.length > 1 && args[0].equals("--secondary")){
			firstHost = new String[args.length - 1];
			for(int i = 1; i < args.length; i++)
				firstHost[i - 1] = args[i];
			isEnded = true;
		}

		return isEnded;
	}

	// ローカルで動かしたいときはこっち
	public static void main(String[] args) throws InterruptedException, IOException{
		ReplicatePeerSystem mainSystem = new ReplicatePeerSystem();

		if(!mainSystem.getArgs(args)) return;

		mainSystem.initialize();

		// FIXME: Msgpack-RPC Netty Log-levelの変更

		mainSystem.console();
	}

	private void console() throws IOException{
		BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
		String arg;

		try{
			Thread.sleep(700);
		}catch(InterruptedException ignored){}

		while(true){
			System.out.print("Fujene > ");
			arg = br.readLine();

			String[] args = arg.split(" ");
			if(args.length > 0){
				if(args[0].equals("index") || args[0].equals("i")){
					if(args.length < 3)
						System.out.println("Usage: index ID <Contents> [<Supplements>]");
					else{
						if(args.length < contentInfo.getContentNum() + contentInfo.getAppendixNum() + 2)
							System.out.println("Input appropriate contents.");
						else setNewIndex(new ExtendQueue(Arrays.copyOfRange(args, 2,
								2 + contentInfo.getContentNum()), Arrays.copyOfRange(args,
								2 + contentInfo.getContentNum(), 2 + contentInfo.getContentNum()
										+ contentInfo.getAppendixNum()), Long.parseLong(args[1])));

					}
				}else if(args[0].equals("search") || args[0].equals("s")){
					if(args.length < 3)
						System.out.println("Specify query.");
					else{
						try{
							ExtendQueue queue;
							int i, j;
							ExtendResult result = setNewSearch(new SearchQueue(args[1], args[2], 0, 100000));
							// ちょっとなら結果を出す
							System.out.println("Result: " + result.getTotal() + " articles.");
							if(result.getLength() <= 10){
								for(i = 0; i < result.getLength(); i++){
									queue = result.getQueue(i);
									System.out.println((i + 1) + ": ID = " + queue.ID);
									for(j = 0; j < queue.contents.length; j++){
										System.out.print(contentInfo.getContentName(j) + " = ");
										System.out.println(queue.contents[j]);
									}
									if(queue.appendices != null && queue.appendices.length > 0)
										System.out.println("---");
									for(j = 0; j < queue.appendices.length; j++){
										System.out.print(contentInfo.getAppendixName(j) + " = ");
										System.out.println(queue.appendices[j]);
									}
								}
							}
						}catch(Exception ignored){
							ignored.printStackTrace();
							continue;
						}
					}
				}else if(args[0].equals("searchspec") || args[0].equals("p")){
					if(args.length != 5)
						System.out.println("Specify arguments.");
					else{
						try{
							ExtendResult result = setNewSearch(new SearchQueue(args[1], args[2],
									Integer.parseInt(args[3]), Integer.parseInt(args[4])));
							int i, j;
							ExtendQueue queue;
							System.out.println("Result total: " + result.getTotal() + " articles, we got "
									+ (Integer.parseInt(args[3]) + 1) + " - "
									+ (Integer.parseInt(args[3]) + result.getLength()));
							if(result.getLength() <= 10){
								for(i = 0; i < result.getLength(); i++){
									queue = result.getQueue(i);
									System.out.println((i + 1 + Integer.parseInt(args[3])) + ": ID = "
											+ queue.ID);
									for(j = 0; j < queue.contents.length; j++){
										System.out.print(contentInfo.getContentName(j) + " = ");
										System.out.println(queue.contents[j]);
									}
									if(queue.appendices != null && queue.appendices.length > 0)
										System.out.println("---");
									for(j = 0; j < queue.appendices.length; j++){
										System.out.print(contentInfo.getAppendixName(j) + " = ");
										System.out.println(queue.appendices[j]);
									}
								}
							}
						}catch(Exception ignored){
							ignored.printStackTrace();
						}
					}
				}else if(args[0].equals("indexcount") || args[0].equals("c")){
					System.out.println("Total count: " + memman.getCounter());
				}else if(args[0].equals("delete") || args[0].equals("d")){
					if(args.length != 2)
						System.out.println("Specify ID.");
					else execIndex.execute(new DeleteHost(Long.parseLong(args[1])));
				}else if(args[0].equals("querycount") || args[0].equals("u")){
					System.out.println("Total count: " + memman.getSearchcounter());
				}else if(args[0].equals("termcount") || args[0].equals("t")){
					System.out.println("Term count: " + memman.getTerms());
				}else if(args[0].equals("skipcount") || args[0].equals("k")){
					System.out.println("Skip ptr. count: " + memman.getSkipPointerLen());
				}else if(args[0].equals("memory") || args[0].equals("m")){
					memman.getMemoryList();
				}else if(args[0].equals("node") || args[0].equals("n")){
					showNodes();
				}else if(args[0].equals("save") || args[0].equals("v")){
					saveMemory();
				}else if(args[0].equals("read") || args[0].equals("r")){
					readMemory();
				}else if(args[0].equals("quit") || args[0].equals("q")){
					setStopper(true);
				}else{
					System.out
							.println("Command: <i>ndex ID <args>, <s>earch <args> type, <d>elete ID, searchs<p>ec <args> from length, <t>ermcount, index<c>ount, q<u>erycount, s<k>ipcount, <m>emory, <n>ode, <r>ead, sa<v>e, <q>uit");
				}
			}
		}
	}

	// 外部からJUnitとかで呼び出すときはこっち
	public void testMain(String[] args) throws InterruptedException, IOException{
		this.blockNum = 32;
		this.blockSize = 1048576;
		this.directory = "./testDir/";
		this.initialize();
	}

	// 全部の動作を止める操作
	public void setStopper(boolean isStop){
		liveThreadOuter.interrupt();
		liveThreadInner.interrupt();

		stopFlag = isStop;
		execIndex.shutdownNow();
		execSearch.shutdownNow();

		try{
			synchronized(this){
				wait(8);
			}
		}catch(InterruptedException e){}
		// 無理やりな気がする
		Runtime.getRuntime().exit(0);
	}

	public void setNewIndex(ExtendQueue newIndex){
		execIndex.execute(new IndexContentsHost(newIndex));
		for(int i = 0; i < newIndex.contents.length; i++){
			TermList terms = chopper.chop(newIndex.contents[i]);
			for(Term term: terms){
				execIndex.execute(new IndexTermHost(term.str(), term.freq(), newIndex.ID, i));
			}
		}
	}

	public class IndexContentsHost implements Runnable{
		private ExtendQueue newIndex;

		public IndexContentsHost(ExtendQueue index){
			this.newIndex = index;
		}

		@Override
		public void run(){
			long startTime;
			long hash;
			Node nextNode;

			hash = ContentHash.getContentHash(newIndex.ID);
			startTime = System.nanoTime();
			nextNode = getNodeByHash(hash, contentNodes);
			mainLogger.finest("Index / Get node: " + ((System.nanoTime() - startTime) / 1000.0)
					+ " microsec.");
			MessagePackObject[] contentObjs = communicator.makeRemoteContentObject(newIndex);

			for(int i = 0; i < (replication > contentNodes.size() ? contentNodes.size() : replication); i++){
				mainLogger.fine("Store content: " + nextNode);
				if(!nextNode.equals(myContentNode)){
					for(int j = 0; j < retryCount; j++){
						try{
							communicator.indexContentRemote(ArrayType.create(contentObjs), newIndex.ID,
									nextNode);
						}catch(Exception e){
							mainLogger.warning("Can not index numbered " + newIndex.ID
									+ " due to server down.");
							mainLogger.warning("Retry to index in next node.");
							nextNode = getNext(nextNode, contentNodes);
							continue;
						}
					}
				}else{
					// インデックス完成まで待ち状態
					if(!myContentNode.isIndexable()) synchronized(myContentNode.indexableSwitch){
						try{
							myContentNode.indexableSwitch.wait();
						}catch(InterruptedException e){}
					}
					startTime = System.nanoTime();
					LookupPointer[] suppPtr = new LookupPointer[newIndex.appendixTypes];
					for(int j = 0; j < newIndex.appendixTypes; j++)
						suppPtr[j] = memman.storePureContent(newIndex.appendices[j], newIndex.ID, j);

					for(int j = 0; j < newIndex.contentTypes; j++)
						memman.storeContent(newIndex.contents[j], newIndex.ID, j, suppPtr);

					mainLogger.finest("Index / Index content: " + ((System.nanoTime() - startTime) / 1000.0)
							+ " microsec.");
				}
				nextNode = getNext(nextNode, contentNodes);
			}
		}
	}

	public class IndexContentsClient implements Runnable{
		private ExtendQueue newIndex;

		public IndexContentsClient(ExtendQueue index){
			this.newIndex = index;
		}

		@Override
		public void run(){
			try{
				if(!myContentNode.isIndexable()) synchronized(myContentNode.indexableSwitch){
					myContentNode.indexableSwitch.wait();
				}
			}catch(InterruptedException e){}
			LookupPointer[] suppPtr = new LookupPointer[newIndex.appendixTypes];
			for(int i = 0; i < newIndex.appendixTypes; i++)
				suppPtr[i] = memman.storePureContent(newIndex.appendices[i], newIndex.ID, i);

			for(int i = 0; i < newIndex.contentTypes; i++)
				memman.storeContent(newIndex.contents[i], newIndex.ID, i, suppPtr);
		}
	}

	public class DeleteHost implements Runnable{
		long ID;

		public DeleteHost(long ID){
			this.ID = ID;
		}

		@Override
		public void run(){
			Node nextNode;

			try{
				long hash = ContentHash.getContentHash(ID);
				nextNode = getNodeByHash(hash, contentNodes);

				for(int i = 0; i < (replication > contentNodes.size() ? contentNodes.size() : replication); i++){
					if(!nextNode.equals(myContentNode)){
						communicator.deleteContent(nextNode, ID);
					}else{
						// インデックス完成まで待ち状態
						if(!myContentNode.isIndexable()) synchronized(myContentNode.indexableSwitch){
							myContentNode.indexableSwitch.wait();
						}

						memman.deleteContent(ID, 0);
					}
					nextNode = getNext(nextNode, contentNodes);
				}
			}catch(Exception e){}
		}
	}

	public class DeleteClient implements Runnable{
		long ID;

		public DeleteClient(long ID){
			this.ID = ID;
		}

		@Override
		public void run(){

			// インデックス完成まで待ち状態
			try{
				if(!myContentNode.isIndexable()) synchronized(myContentNode.indexableSwitch){
					myContentNode.indexableSwitch.wait();
				}
			}catch(InterruptedException e){}

			memman.deleteContent(ID, 0);
		}
	}

	public class IndexTermHost implements Runnable{
		private String newTerm;
		private long termHash;
		private long ID;
		private int valueType;
		private int termfreq;

		public IndexTermHost(String newTerm, int termfreq, long ID, int valueType){
			this.newTerm = newTerm;
			this.termHash = ContentHash.getTermHash(newTerm);
			this.ID = ID;
			this.valueType = valueType;
			this.termfreq = termfreq;
		}

		@Override
		public void run(){
			Node nextNode = getNodeByHash(termHash, termNodes);

			for(int i = 0; i < (replication > termNodes.size() ? termNodes.size() : replication); i++){
				if(nextNode.equals(myTermNode)){
					try{
						if(!myTermNode.isIndexable()) synchronized(myTermNode.indexableSwitch){
							myTermNode.indexableSwitch.wait();
						}
					}catch(InterruptedException e){}

					memman.storeTerm(newTerm, ID, termfreq, valueType);
				}else{
					for(int j = 0; j < retryCount; j++){
						try{
							communicator.indexTermRemote(newTerm, termfreq, ID, valueType, nextNode);
						}catch(Exception e){
							mainLogger.warning("Can not index term " + newTerm + " due to server down.");
							mainLogger.warning("Retry to index in next node.");
							nextNode = getNext(nextNode, termNodes);
							continue;
						}
					}
				}
				nextNode = getNext(nextNode, termNodes);
			}
		}
	}

	public class IndexTermClient implements Runnable{
		private String newTerm;
		private long ID;
		private int valueType;
		private int termfreq;

		public IndexTermClient(String newTerm, int termfreq, long ID, int valueType){
			this.newTerm = newTerm;
			this.ID = ID;
			this.valueType = valueType;
			this.termfreq = termfreq;
		}

		@Override
		public void run(){
			try{
				if(!myTermNode.isIndexable()) synchronized(myTermNode.indexableSwitch){
					myTermNode.indexableSwitch.wait();
				}
			}catch(InterruptedException e){}
			memman.storeTerm(newTerm, ID, termfreq, valueType);
		}
	}

	private CountDownLatch searchGate;
	private ExtendTerm[] resultPtrs;
	private long[] termHashes;
	private boolean isNotFound;

	// 大元のsearch窓口
	public ExtendResult setNewSearch(SearchQueue newQuery) throws Exception{
		ExtendResult result = new ExtendResult(contentInfo);

		newQuery.typenum = contentInfo.getContentIndex(newQuery.typestr);

		TermList terms = chopper.chop(newQuery.query);
		searchGate = new CountDownLatch((terms.size() > 10 ? 10 : terms.size()));
		resultPtrs = new ExtendTerm[(terms.size() > 10 ? 10 : terms.size())];
		termHashes = new long[(terms.size() > 10 ? 10 : terms.size())];
		isNotFound = false;
		long time = System.nanoTime();

		int count = 0;
		for(Term term: terms){
			// 処理を投げる
			execSearch.execute(new SearchHost(term.str(), count, newQuery.typenum));
			if(++count > 10) break;
		}
		mainLogger
				.finest("Search / Sending process: " + ((System.nanoTime() - time) / 1000.0) + " microsec.");
		time = System.nanoTime();
		// 全部が検索し終わったらインターセクション
		searchGate.await();
		mainLogger.finest("Search / Search terms: " + ((System.nanoTime() - time) / 1000.0) + " microsec.");

		if(!isNotFound){
			try{
				time = System.nanoTime();
				// インターセクション
				ArrayList<HashIDExt> hashes;
				if(terms.size() > 1){
					hashes = intersection(newQuery.typenum);
					mainLogger.finest("Search / Intersection: " + ((System.nanoTime() - time) / 1000.0)
							+ " microsec.");
					time = System.nanoTime();
					result.setTotal(hashes.size());
					// 本文抽出
					result.setResult(getContents(hashes, newQuery.from, newQuery.length, newQuery.typenum));
				}else{
					// 1単語なら早めに出せるはず
					hashes = getIndexAlone(newQuery.typenum, newQuery.from, newQuery.length);
					mainLogger.finest("Search / Get list: " + ((System.nanoTime() - time) / 1000.0)
							+ " microsec.");
					time = System.nanoTime();
					result.setTotal(resultPtrs[0].docfreq);
					// 本文抽出
					result.setResult(getContents(hashes, 0, hashes.size(), newQuery.typenum));
				}
				if(isRerank) rerank(result);
				mainLogger.finest("Search / Get content: " + ((System.nanoTime() - time) / 1000.0)
						+ " microsec.");
			}catch(Exception e){
				e.printStackTrace();
			}
		}else{
			System.out.println("Not found.");
			result.clearResult();
		}

		return result;
	}

	// 検索サービス
	public class SearchHost implements Runnable{
		private String newQuery = null;
		private int cnt;
		private int valueType;

		public SearchHost(String newQuery, int count, int valueType){
			this.newQuery = newQuery;
			this.cnt = count;
			this.valueType = valueType;
		}

		public void run(){
			try{
				long termHash = ContentHash.getTermHash(newQuery);
				termHashes[cnt] = termHash;
				Node termNode = getNodeByHash(termHash, termNodes);

				mainLogger.fine("Search - Node: " + termNode.toString());
				if(termNode.equals(myTermNode)){
					if(!myTermNode.isIndexable()) synchronized(myTermNode.indexableSwitch){
						myTermNode.indexableSwitch.wait();
					}
					resultPtrs[cnt] = memman.searchTerm(newQuery, valueType);
				}else{
					byte[] extb = null;
					for(int i = 0; i < retryCount; i++){
						try{
							extb = communicator.searchRemote(newQuery, valueType, termNode);
							if(extb != null) resultPtrs[cnt] = new ExtendTerm(extb);
						}catch(Exception e){
							mainLogger.warning("Retrying search in next node.");
							termNode = getNext(termNode, termNodes);

						}
					}
				}
				// 検索結果が出たらカウントダウン
				if(resultPtrs[cnt] == null) isNotFound = true;
			}catch(InterruptedException e){}finally{
				searchGate.countDown();
			}
		}
	}

	// 一度に読み込むHashIDの数
	private static final int BUFFERSIZE = 100;

	/**
	 * 転置インデックスを<code>length</code>だけ読み出します。
	 * @param hash ... 単語ハッシュ
	 * @param fromPtr ... 転置インデックス開始位置
	 * @param length ... 長さ
	 * @return
	 */
	private LinkedHashMap<Long, Integer> getContentList(long hash, LookupPointer fromPtr, int length){
		LinkedHashMap<Long, Integer> hashMap = null;

		try{
			Node termNode = getNodeByHash(hash, termNodes);
			if(termNode.equals(myTermNode))
				hashMap = memman.getContentsByHash(fromPtr, length);
			else hashMap = communicator.getInvertIndexList(termNode, fromPtr);

			return hashMap;
		}catch(Exception e){
			e.printStackTrace();
			return null;
		}
	}

	/**
	 * 転置インデックスをすべて取り出します。
	 * @param hash
	 * @param fromPtr
	 * @return
	 */
	private TreeMap<Long, Integer> getAllContentList(long hash, LookupPointer fromPtr){
		TreeMap<Long, Integer> retList = new TreeMap<Long, Integer>(new Comparator<Long>(){
			@Override
			public int compare(Long o1, Long o2){
				return (o1.longValue() < o2.longValue() ? 1 : (o1.longValue() > o2.longValue() ? -1 : 0));
			}
		});

		LinkedHashMap<Long, Integer> pieceMap;

		while(!fromPtr.equals(Constants.EMPTY_PTR)){
			pieceMap = getContentList(hash, fromPtr, BUFFERSIZE);
			if(pieceMap != null){
				/*
				 * for(Entry<Long, Integer> entry: pieceMap.entrySet())
				 * retList.put(entry.getKey(), entry.getValue());
				 */
				retList.putAll(pieceMap);
			}
		}

		return retList;
	}

	/**
	 * 1単語のみインデックスをとります。
	 * @param prior ... 検索対象インデックス
	 * @param from ... 検索結果開始位置
	 * @param length ... 検索結果長さ
	 * @return ... <code>HashID</code>のリスト
	 */
	private ArrayList<HashIDExt> getIndexAlone(int prior, int from, int length){
		ArrayList<HashIDExt> results = new ArrayList<HashIDExt>(length + 1);
		resultPtrs[0].invIndex[prior].firstID = resultPtrs[0].firstIDs[prior];
		LinkedHashMap<Long, Integer> IDMap = getContentList(termHashes[0], resultPtrs[0].invIndex[prior],
				from + length);
		double tfidf;
		int count = 0;

		for(Entry<Long, Integer> eachID: IDMap.entrySet()){
			if(count < from || count >= from + length) continue;
			tfidf = (eachID.getValue()).doubleValue() / resultPtrs[0].docfreq;
			results.add(new HashIDExt(ContentHash.getContentHash((eachID.getKey()).longValue()), (eachID
					.getKey()).longValue(), tfidf));
		}

		return results;
	}

	/**
	 * 複数単語のインデックスの共通部分(intersection)をとります。
	 * @param prior
	 * @return
	 */
	@SuppressWarnings("unchecked")
	private ArrayList<HashIDExt> intersection(int prior){
		ArrayList<HashIDExt> results = new ArrayList<HashIDExt>();

		int ptrlen = resultPtrs.length;
		long[] currentPtrs = new long[ptrlen];
		Iterator<Entry<Long, Integer>>[] iters = new Iterator[ptrlen];
		Entry<Long, Integer>[] entries = new Entry[ptrlen];
		TreeMap<Long, Integer>[] sortedEntries = new TreeMap[ptrlen];
		int[] currentNum = new int[ptrlen];
		int counter = 0;
		boolean samePtr = false;
		boolean isDone = false;
		int newestIndex;
		long newestPtr;
		long comps;
		int i;

		// 計算量: O(n)
		for(i = 0; i < ptrlen; i++){
			resultPtrs[i].invIndex[prior].firstID = resultPtrs[i].firstIDs[prior];
			sortedEntries[i] = getAllContentList(termHashes[i], resultPtrs[i].invIndex[prior]);
			iters[i] = sortedEntries[i].entrySet().iterator();
			entries[i] = iters[i].next();
			currentPtrs[i] = (entries[i].getKey()).longValue();
			currentNum[i] = 0;
		}

		while(counter < 100000){
			samePtr = true;
			newestIndex = 0;
			newestPtr = currentPtrs[0];
			// ポインタの比較
			for(i = 1; i < ptrlen; i++){
				comps = newestPtr - currentPtrs[i];
				if(comps < 0){
					newestIndex = i;
					newestPtr = currentPtrs[i];
					samePtr = false;
				}else if(samePtr && comps > 0) samePtr = false;
			}

			if(samePtr){
				// 一致 全ポインタ進め
				double tfidf = 0.0;
				for(i = 0; i < ptrlen; i++)
					tfidf += (entries[i].getValue()).doubleValue() / resultPtrs[i].docfreq;

				results.add(new HashIDExt(ContentHash.getContentHash(newestPtr), newestPtr, tfidf));
				counter++;
				for(i = 0; i < ptrlen; i++){
					// もうない
					if(!iters[i].hasNext()){
						isDone = true;
						break;
					}
					entries[i] = iters[i].next();
					currentPtrs[i] = entries[i].getKey().longValue();
				}
			}else{
				// 一番新しいものを進める

				currentNum[newestIndex]++;
				// もうない
				if(!iters[newestIndex].hasNext()){
					isDone = true;
					break;
				}
				entries[newestIndex] = iters[newestIndex].next();
				currentPtrs[newestIndex] = entries[newestIndex].getKey().longValue();
			}

			if(isDone) break;
		}

		return results;
	}

	public ExtendQueue[] getContents(ArrayList<HashIDExt> hash, int from, int length, int priorIndex){
		ExtendQueue[] result = new ExtendQueue[(length > hash.size() ? hash.size() : length)];
		Node lookupNode;
		HashIDExt eachHash;

		for(int i = 0; i < hash.size(); i++){
			if(i >= from && i < from + length){
				eachHash = hash.get(i);
				lookupNode = getNodeByHash(eachHash.hash, contentNodes);
				if(lookupNode.equals(myContentNode)){
					result[i - from] = memman.lookupContent(eachHash.ID, priorIndex);
					if(result[i - from].equals(Constants.EMPTY_QUEUE))
						i--;
					else result[i - from].score = eachHash.score;
				}else{
					result[i - from] = communicator.getContentSet(eachHash.ID, priorIndex,
							contentInfo.getContentNum(), contentInfo.getAppendixNum(), lookupNode);
					result[i - from].score = eachHash.score;
				}
			}
		}

		return result;
	}

	public void readMemory(){
		memman.initializeMemory();
	}

	public void saveMemory(){
		memman.saveMemory();
	}

	// Re-rank by score
	private void rerank(ExtendResult result){
		TreeSet<ExtendQueue> reranked = new TreeSet<ExtendQueue>(new Comparator<ExtendQueue>(){
			@Override
			public int compare(ExtendQueue q1, ExtendQueue q2){
				return (q1.score > q2.score ? 1 : (q1.score < q2.score ? -1 : 0));
			}
		});

		for(int i = 0; i < result.getQueues().length; i++){
			ExtendQueue queue = result.getQueue(i);
			if(queue != null) reranked.add(queue);
		}

		result.setResult(reranked.toArray(new ExtendQueue[0]));
	}

	// ///////////////////// P2P関連の実装 //////////////////////
	// ノードを返します
	public TreeSet<Node> getNode(int type){
		if(type == OUTER)
			return termNodes;
		else if(type == INNER)
			return contentNodes;
		else return null;
	}

	// インデックスを投げます
	public void invokeIndexInner(ExtendQueue queue){
		execIndex.execute(new IndexContentsClient(queue));
	}

	// インデックスを投げます
	public void invokeIndexOuter(String str, int termfreq, long ID, int valueType){
		execIndex.execute(new IndexTermClient(str, termfreq, ID, valueType));
	}

	public ExtendTerm invokeSearch(String query, int valueType){
		try{
			if(!myTermNode.isIndexable()) synchronized(myTermNode.indexableSwitch){
				myTermNode.indexableSwitch.wait();
			}
		}catch(InterruptedException e){}
		return memman.searchTerm(query, valueType);
	}

	public void invokeDelete(long ID){
		execIndex.execute(new DeleteClient(ID));
	}

	public ArrayList<ExtendQueue> getContentListByHash(long hashFrom, long hashTo, int valueType){
		return memman.getContentList(hashFrom, hashTo, valueType);
	}

	public Future<ArrayList<ExtendQueue>> getContentSnapshot(long hashFrom, long hashTo, int valueType){
		ExecutorService snapperService = Executors.newCachedThreadPool();

		return snapperService.submit(new ContentSnapshoter(memman, hashFrom, hashTo, valueType));
	}

	public ArrayList<ExtendTerm> getTermListByHash(long hashFrom, long hashTo, int valueType){
		return memman.getTermList(hashFrom, hashTo, valueType);
	}

	public Future<ArrayList<ExtendTerm>> getTermSnapshot(long hashFrom, long hashTo, int valueType){
		ExecutorService snapperService = Executors.newCachedThreadPool();

		return snapperService.submit(new TermSnapshoter(memman, hashFrom, hashTo, valueType));
	}

	public LinkedHashMap<Long, Integer> getContentsByHash(LookupPointer ptr, int length){
		return memman.getContentsByHash(ptr, length);
	}

	public Future<LinkedHashMap<Long, Integer>> getInvIndexSnapshot(LookupPointer ptr, int length){
		ExecutorService snapperService = Executors.newCachedThreadPool();

		return snapperService.submit(new InvertIndexSnapshoter(memman, ptr, length));
	}

	public ArrayType putInfo(){
		MessagePackObject[] obj = new MessagePackObject[7 + contentInfo.getContentNum()
				+ contentInfo.getAppendixNum()];
		obj[0] = IntegerType.create(blockNum);
		obj[1] = IntegerType.create(blockSize);
		obj[2] = IntegerType.create(replication);
		obj[3] = IntegerType.create(indexType);
		obj[4] = IntegerType.create(contentInfo.getContentNum());
		obj[5] = IntegerType.create(contentInfo.getAppendixNum());
		obj[6] = RawType.create(contentInfo.getConnection());
		for(int i = 0; i < contentInfo.getContentNum(); i++)
			obj[7 + i] = RawType.create(contentInfo.getContentName(i));
		for(int i = 0; i < contentInfo.getAppendixNum(); i++)
			obj[7 + contentInfo.getContentNum() + i] = RawType.create(contentInfo.getAppendixName(i));

		return ArrayType.create(obj);
	}

	public int getQueueSize(int type){
		if(type == 0)
			return ((ThreadPoolExecutor)execIndex).getPoolSize();
		else if(type == 1)
			return ((ThreadPoolExecutor)execSearch).getPoolSize();
		else return -1;
	}

	public ExtendQueue lookupContent(long ID, int index){
		return memman.lookupContent(ID, index);
	}

	// 自分のハッシュ値を計算する
	private long[] calcHash(InetAddress addrInfo){
		// OuterとInner用に
		long[] returnHash = new long[2];

		// とりあえずマシン名で作る
		returnHash[0] = addrInfo.getHostName().hashCode() * addrInfo.getHostName().hashCode()
				* 0x2468ACE2468ACE23L;
		Long.rotateLeft(returnHash[0], 45);
		// とりあえずかき混ぜる
		returnHash[1] = returnHash[0] ^ 0xAAAAAAAAAAAAAAAAL * 0x2468ACE2468ACE23L;
		Long.rotateLeft(returnHash[1], 31);

		return returnHash;
	}

	// n個前(hashが小さい方)のノードを取得
	private Node getDescentHash(Node from, int distance, TreeSet<Node> nodes){
		if(from == null) return null;
		Node getNode = from;
		for(int i = 0; i < distance; i++)
			getNode = getPrev(getNode, nodes);

		return getNode;
	}

	// 次のノードを取得
	private Node getNext(Node from, TreeSet<Node> nodes){
		if(from == null) return null;
		TreeSet<Node> calcNode = new TreeSet<Node>(nodes);
		boolean isNext = false;

		for(Node eachNode: calcNode){
			if(isNext) return eachNode;
			if(eachNode.equals(from)) isNext = true;
		}

		return calcNode.first();
	}

	// 前のノードを取得
	private Node getPrev(Node from, TreeSet<Node> nodes){
		if(from == null) return null;
		TreeSet<Node> calcNode = new TreeSet<Node>(nodes);
		Node prevNode = null;

		for(Node eachNode: calcNode){
			if(eachNode.equals(from)) if(prevNode == null)
				return calcNode.last();
			else return prevNode;
			prevNode = eachNode;
		}

		return null;
	}

	/**
	 * ハッシュ直後のノードを取得します。
	 * @param hash
	 * @param nodelist
	 * @return
	 */
	private Node getNodeByHash(long hash, TreeSet<Node> nodelist){
		Node currentPeer = new Node("Dummy", hash);

		if(nodelist.ceiling(currentPeer) != null)
			return nodelist.ceiling(currentPeer);
		else return nodelist.first();
	}

	// IPアドレスからノードを取得します。
	private Node getNodeByAddr(String address, TreeSet<Node> nodelist){
		TreeSet<Node> copylist = new TreeSet<Node>(nodelist);

		for(Node node: copylist)
			if(node.getAddr().equals(address)) return node;

		return null;
	}

	// ノード間がチェーン内でいくつ離れているか取得 0はエラー
	private int getDistance(Node from, Node to, TreeSet<Node> nodes){
		// 2つのノードがあるか確認
		if(!nodes.contains(from) || !nodes.contains(to)) return 0;

		Node nextPeer = getNext(from, nodes);
		Node prevPeer = getPrev(from, nodes);
		for(int i = 1;; i++){
			if(nextPeer.equals(to)) return i;
			if(prevPeer.equals(to)) return (-1 * i);
			nextPeer = getNext(nextPeer, nodes);
			prevPeer = getPrev(prevPeer, nodes);
		}
	}

	// サークル情報を取得します。
	private void getInformation(){
		if(firstHost != null){
			for(String hosts: firstHost){
				ReplicateInfo repInfo = communicator.getInfo(hosts);
				if(repInfo == null) continue; // 情報がもらえなければ他のノードから
				this.blockNum = repInfo.blockNum;
				this.blockSize = repInfo.blockSize;
				this.replication = repInfo.replication;
				this.indexType = repInfo.indexType;
				this.contentInfo = repInfo.contentInfo;
			}
		}
	}

	// 追加操作を行います。
	public void addReplication(int type, Node newPeer){
		// サークルに追加
		if(type == OUTER)
			termNodes.add(newPeer);
		else if(type == INNER) contentNodes.add(newPeer);
	}

	// ノードを登録します。
	private void registerNode(){
		for(String host: firstHost){
			// ノードリスト要求
			try{
				termNodes = communicator.getNodeList(host, OUTER);
				termNodes.add(myTermNode);
				contentNodes = communicator.getNodeList(host, INNER);
				contentNodes.add(myContentNode);
				if(termNodes == null || contentNodes == null) continue;
			}catch(Exception e){
				mainLogger.warning("Node might be not found, retrying in next node.");
				continue;
			}

			// 現時点でのノードリストを使って自分のことを送信
			TreeSet<Node> currentNode = new TreeSet<Node>(termNodes);
			for(Node eachNode: currentNode){
				if(!eachNode.equals(myTermNode))
					communicator.register(eachNode.getAddr(), myTermNode.getAddr(), myTermNode.getHash(),
							myContentNode.getHash());
			}
			break;
		}
	}

	// データ要求をします。
	private void callForData(int type){
		TreeSet<Node> nodes;
		long hashFrom, hashTo;
		Node myNode;
		int count = 0;

		if(type == OUTER){
			nodes = termNodes;
			myNode = myTermNode;
		}else if(type == INNER){
			nodes = contentNodes;
			myNode = myContentNode;
		}else return;

		// ハッシュ範囲の計算
		if(nodes.size() > replication){
			hashFrom = getDescentHash(myNode, replication, nodes).getHash();
			hashTo = myNode.getHash();
		}else{
			// ぜんぶ
			hashFrom = 0;
			hashTo = 0xFFFFFFFFFFFFFFFFL;
		}

		Node nextNode = getNext(myNode, nodes);

		if(type == OUTER){
			HashMap<String, ArrayList<LookupPointer>> termMap = communicator.getTermList(nextNode, hashFrom,
					hashTo, myNode.getAddr());
			LinkedHashMap<Long, Integer> invIndices = null;
			for(Entry<String, ArrayList<LookupPointer>> eachTerm: termMap.entrySet()){
				for(int i = 0; i < eachTerm.getValue().size(); i++){
					invIndices = communicator.getInvertIndexList(nextNode, eachTerm.getValue().get(i));
					for(Entry<Long, Integer> index: invIndices.entrySet())
						execIndex.execute(new IndexTermClient(eachTerm.getKey(), index.getValue().intValue(),
								index.getKey().longValue(), i));
				}
				if(count++ % 500 == 0 || count == termMap.size()){
					String sharps = " ";
					for(int j = 0; j < (count * 48 / termMap.size()); j++)
						sharps = sharps.concat("#");
					for(int j = sharps.length(); j <= 48; j++)
						sharps = sharps.concat(" ");

					System.out.printf("\r%3d pct.%s %8d", (count * 100 / termMap.size()), sharps, count);
				}

			}
		}else if(type == INNER){
			HashSet<ExtendQueue> contentSet = communicator.getContentListFromNode(nextNode, hashFrom, hashTo,
					myNode.getAddr(), contentInfo);
			for(ExtendQueue eachQueue: contentSet){
				execIndex.execute(new IndexContentsClient(eachQueue));
				if(count++ % 500 == 0 || count == contentSet.size()){
					String sharps = " ";
					for(int j = 0; j < (count * 48 / contentSet.size()); j++)
						sharps = sharps.concat("#");
					for(int j = sharps.length(); j <= 48; j++)
						sharps = sharps.concat(" ");

					System.out.printf("\r%3d pct.%s %8d", (count * 100 / contentSet.size()), sharps, count);
				}
			}
		}
		System.out.println();

		// indexableにして上書き
		myNode.indexable();
	}

	// 削除操作を行います。
	public void deleteReplication(int type, Node delPeer){
		Node myNode;
		TreeSet<Node> localNodeList;
		if(type == OUTER){
			localNodeList = termNodes;
			myNode = myTermNode;
		}else{
			localNodeList = contentNodes;
			myNode = myContentNode;
		}
		int dist = getDistance(myNode, delPeer, localNodeList);
		if(localNodeList.size() > replication && dist >= 1 && dist <= replication){
			// レプリケーション維持に必要な範囲を決定
			Node dataNode;
			long hashTo;

			// 削除ノードをよけてコピーできるノードを分散させる
			if(dist == replication){
				dataNode = getNext(myNode, localNodeList);
				hashTo = delPeer.getHash();
			}else{
				dataNode = getDescentHash(myNode, replication - 1, localNodeList);
				hashTo = dataNode.getHash();
			}
			long hashFrom = getPrev(dataNode, localNodeList).getHash();

			// チェーンから削除
			localNodeList.remove(delPeer);

			if(type == OUTER){
				HashMap<String, ArrayList<LookupPointer>> termMap = communicator.getTermList(dataNode,
						hashFrom, hashTo, myNode.getAddr());
				LinkedHashMap<Long, Integer> invIndices = null;
				for(Entry<String, ArrayList<LookupPointer>> eachTerm: termMap.entrySet()){
					for(int i = 0; i < eachTerm.getValue().size(); i++){
						invIndices = communicator.getInvertIndexList(dataNode, eachTerm.getValue().get(i));
						for(Entry<Long, Integer> index: invIndices.entrySet()){
							execIndex.execute(new IndexTermClient(eachTerm.getKey(), index.getValue()
									.intValue(), index.getKey().longValue(), i));
						}
					}
				}
			}else if(type == INNER){
				HashSet<ExtendQueue> contentSet = communicator.getContentListFromNode(dataNode, hashFrom,
						hashTo, myNode.getAddr(), contentInfo);
				for(ExtendQueue eachQueue: contentSet)
					execIndex.execute(new IndexContentsClient(eachQueue));
			}

		}else{
			// ノードリストから削除
			localNodeList.remove(delPeer);
			if(type == OUTER)
				termNodes = localNodeList;
			else contentNodes = localNodeList;

			System.out.println("Node number: " + termNodes.size() + ", " + contentNodes.size());
		}
	}

	// 自分の制御さえできれば十分
	public void iamIndexable(){
		if(!myTermNode.isIndexable()) myTermNode.indexable();
		if(!myContentNode.isIndexable()) myContentNode.indexable();
	}

	public void stopIndexable(){
		if(myTermNode.isIndexable()) myTermNode.notIndexable();
		if(myContentNode.isIndexable()) myContentNode.notIndexable();
	}

	private void showNodes(){
		for(int i = OUTER; i <= INNER; i++){
			System.out.println((i == OUTER ? "--- OUTER ---" : "--- INNER ---"));
			TreeSet<Node> showNode = new TreeSet<Node>((i == OUTER ? termNodes : contentNodes));
			int count = 0;

			for(Node eachNode: showNode){
				System.out.println((count++) + ": Hash-" + String.format("0x%016x", eachNode.getHash())
						+ " Address-" + eachNode.getAddr() + " Indexable-"
						+ (eachNode.isIndexable() ? "Yes" : "No"));
			}
		}
	}

	/**
	 * ピアの生存確認をします。
	 */
	public class SurvivingService extends Thread{
		private int type;

		public SurvivingService(int type){
			this.type = type;
		}

		@Override
		public void run(){
			while(true){
				TreeSet<Node> infos = new TreeSet<Node>(type == OUTER ? termNodes : contentNodes);
				boolean isLiveSignal;

				for(Node eachInfo: infos){
					if((type == OUTER && !eachInfo.equals(myTermNode))
							|| (type == INNER && !eachInfo.equals(myContentNode))){
						try{
							isLiveSignal = communicator.isLive(eachInfo);
							if(!isLiveSignal)
								// クライアント死亡と推定
								deleteReplication(type, eachInfo);
							else if(stopFlag)
							// 自分が死ぬことを通知
								communicator.iWillDie(eachInfo);
						}catch(Exception e){
							mainLogger.info("Node " + eachInfo.getAddr() + " seems to be dead.");
							// クライアント死亡と推定
							deleteReplication(type, eachInfo);
						}
					}
				}

				if(stopFlag) break;
				try{
					TimeUnit.SECONDS.sleep(60);
				}catch(InterruptedException e){
					continue;
				}
			}
		}
	}
}
